package es.hekorusoft.udpcache.server;

import java.io.IOException;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.HashMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import net.sf.ehcache.Cache;
import net.sf.ehcache.CacheManager;
import net.sf.ehcache.Element;
import net.sf.ehcache.Statistics;
import es.hekorusoft.udpcache.messages.ClientMessage;
import es.hekorusoft.udpcache.messages.ServerMessage;

/*
 This file is part of UDPCache.

 UDPCache is free software: you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation, either version 3 of the License, or
 (at your option) any later version.

 UDPCache is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with UDPCache.  If not, see <http://www.gnu.org/licenses/>.
 */
public class UDPLayer {

	private final CacheManager cacheManager;
	private final DatagramChannel datagramChannel;
	private final BlockingQueue<MessageTask> messageQueue;
	private final InetSocketAddress serverAddress;
	private final BlockingQueue<CacheRequest> requestQueue;
	private final HashMap<String, Cache> cacheMap;

	public UDPLayer(int port, boolean deserialize, boolean daemon) throws IOException {

		// Cache manager
		this.cacheManager = CacheManager.create();
		this.cacheMap = new HashMap<String, Cache>(cacheManager.getCacheNames().length);

		// IO
		this.serverAddress = new InetSocketAddress("localhost", port);
		datagramChannel = DatagramChannel.open();
		DatagramSocket socket = datagramChannel.socket();
		socket.bind(serverAddress);

		// Queues
		this.messageQueue = new LinkedBlockingQueue<MessageTask>();
		this.requestQueue = new LinkedBlockingQueue<CacheRequest>();

		// SEND Thread
		Thread senderThread = new Thread(new UDPSender());
		senderThread.setName("UDPSender Thread");
		senderThread.setDaemon(daemon);
		senderThread.setPriority(Thread.MAX_PRIORITY);

		// RCV Thread
		Thread receiverThread = new Thread(new UDPReceiver());
		receiverThread.setName("UDPReceiver Thread");
		receiverThread.setDaemon(daemon);
		receiverThread.setPriority(Thread.MAX_PRIORITY);

		receiverThread.start();
		senderThread.start();

		// Cache workers
		CacheWorker cacheWorker = new CacheWorker();
		int numberOfCpus = Runtime.getRuntime().availableProcessors() + 1;

		System.out.println("Number of threads: " + numberOfCpus);

		for (int i = 0; i < numberOfCpus; i++) {
			Thread cacheWorkerThread = new Thread(cacheWorker);
			cacheWorkerThread.setName("CacheWorker #" + i);
			cacheWorkerThread.setDaemon(daemon);
			cacheWorkerThread.start();
		}

		System.out.println("Inicializado! " + serverAddress);
	}

	private final class CacheWorker implements Runnable {

		public void run() {
			for (;;) {
				try {
					final CacheRequest request = requestQueue.take();
					final ClientMessage msg = ClientMessage.fromBytes(request.copia);

					final String cacheName = msg.getCache();
					Cache cache = cacheMap.get(cacheName);
					if (cache == null) {

						if (!cacheManager.cacheExists(cacheName)) {
							cacheManager.addCache(cacheName);
						}
						cache = cacheManager.getCache(cacheName);
						cacheMap.put(cacheName, cache);
					}

					// Cache cache = cacheManager.getCache(cacheName);

					if (cache != null) {
						switch (msg.getCode()) {
						case GET: {
							Element cacheElement = cache.get(msg.getKey());
							ServerMessage smsg;
							if (cacheElement != null) {
								smsg = ServerMessage.foundMessage(msg.getSequence(), (byte[]) cacheElement.getObjectValue(), cacheElement
										.getExpirationTime());
							} else {
								smsg = ServerMessage.notFoundMessage(msg.getSequence());
							}

							messageQueue.add(new MessageTask(request.socketAddress, smsg));
							break;
						}
						case PUT: {
							cache.put(new Element(msg.getKey(), msg.getByteObj()));
							break;
						}
						case REMOVE: {
							cache.remove(msg.getKey());
							break;
						}
						case REMOVEALL: {
							cache.removeAll();
							break;
						}
						case STATS: {
							Statistics cacheStats = cache.getStatistics();
							messageQueue.add(new MessageTask(request.socketAddress, ServerMessage.cacheStatsMessage(msg.getSequence(), cacheStats)));
							break;
						}
						}
					}
				} catch (Exception ex) {
					ex.printStackTrace();
				}
			}

		}
	}

	private static final class MessageTask {

		private MessageTask(SocketAddress address, ServerMessage msg) {
			this.address = address;
			this.msg = msg;
		}

		private SocketAddress address;
		private ServerMessage msg;
	}

	public final class UDPReceiver implements Runnable {

		public void run() {
			ByteBuffer buffer = ByteBuffer.allocateDirect(1024 * 63);

			for (;;) {

				try {
					SocketAddress address = datagramChannel.receive(buffer);
					int leidos = buffer.position();
					byte[] copia = new byte[leidos];

					buffer.rewind();
					buffer.get(copia);
					requestQueue.add(new CacheRequest(copia, address));
					buffer.clear();

				} catch (IOException e) {
					e.printStackTrace();
				}
			}

		}

	}

	public final class UDPSender implements Runnable {

		public void run() {
			try {

				for (;;) {

					MessageTask task = messageQueue.take();
					datagramChannel.send(ByteBuffer.wrap(task.msg.toByteArray()), task.address);

				}
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		}

	}

	private final static class CacheRequest {

		private final byte[] copia;
		private final SocketAddress socketAddress;

		private CacheRequest(byte[] copia, SocketAddress socketAddress) {
			this.copia = copia;
			this.socketAddress = socketAddress;
		}

	}

	public static void main(String... args) throws Exception {
		new UDPLayer(20000, false, false);
		// Thread.currentThread().sleep(1000000000);

	}

}